//! Tokio网络层实现
//! 
//! 核心设计：
//! 1. 网络层使用Tokio异步处理I/O
//! 2. 通过标准库的mpsc与协议层通信
//! 3. 协议层完全同步，不依赖Tokio
//! 4. 完美的关注点分离

use std::sync::{Arc, Mutex};
use std::collections::HashMap;
use std::io;
use tokio::net::{TcpListener, TcpStream};
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::sync::mpsc as tokio_mpsc;
use tokio::time::{sleep, Duration};

use crate::protocol::{
    DecoupledConnectionHandler, 
    DecoupledProtocolServer
};
// 注意：不再导入NetworkStream，因为Tokio网络层不需要实现协议层的抽象接口
use crate::protocol::decoupled_connection_handler::ConnectionResponse;
use crate::storage::SingleThreadCommandDispatcher;
use crate::event::ConnectionId;

// 注意：这个文件是纯Tokio网络层实现
// 不需要实现NetworkStream trait，因为协议层已经完全解耦
// 协议层使用DecoupledConnectionHandler，通过标准库channel通信

/// Tokio网络服务器
/// 
/// 这是网络层的实现，使用Tokio处理异步I/O
/// 与协议层通过标准库的channel通信
pub struct TokioNetworkServer {
    /// 协议服务器
    protocol_server: Arc<DecoupledProtocolServer>,
    /// 活跃连接的发送器
    connection_senders: Arc<Mutex<HashMap<ConnectionId, tokio_mpsc::UnboundedSender<Vec<u8>>>>>,
    /// 是否运行中
    running: Arc<tokio::sync::RwLock<bool>>,
}

impl TokioNetworkServer {
    /// 创建新的Tokio网络服务器
    pub fn new(storage_dispatcher: SingleThreadCommandDispatcher) -> Self {
        Self {
            protocol_server: Arc::new(DecoupledProtocolServer::new(storage_dispatcher)),
            connection_senders: Arc::new(Mutex::new(HashMap::new())),
            running: Arc::new(tokio::sync::RwLock::new(false)),
        }
    }

    /// 启动服务器
    pub async fn start(&self, bind_addr: &str) -> Result<(), String> {
        {
            let mut running = self.running.write().await;
            if *running {
                return Err("服务器已在运行".to_string());
            }
            *running = true;
        }

        // 绑定TCP监听器
        let listener = TcpListener::bind(bind_addr).await
            .map_err(|e| format!("绑定地址 {} 失败: {}", bind_addr, e))?;

        println!("🚀 Tokio网络服务器启动: {}", bind_addr);

        // 启动接受连接的任务
        let protocol_server = Arc::clone(&self.protocol_server);
        let connection_senders = Arc::clone(&self.connection_senders);
        let running = Arc::clone(&self.running);

        tokio::spawn(async move {
            Self::accept_connections(listener, protocol_server, connection_senders, running).await;
        });

        Ok(())
    }

    /// 停止服务器
    pub async fn stop(&self) -> Result<(), String> {
        {
            let mut running = self.running.write().await;
            if !*running {
                return Ok(());
            }
            *running = false;
        }

        // 清理所有连接
        {
            let mut senders = self.connection_senders.lock().unwrap();
            senders.clear();
        }

        println!("🛑 Tokio网络服务器已停止");
        Ok(())
    }

    /// 接受连接的主循环
    async fn accept_connections(
        listener: TcpListener,
        protocol_server: Arc<DecoupledProtocolServer>,
        connection_senders: Arc<Mutex<HashMap<ConnectionId, tokio_mpsc::UnboundedSender<Vec<u8>>>>>,
        running: Arc<tokio::sync::RwLock<bool>>,
    ) {
        while *running.read().await {
            match listener.accept().await {
                Ok((stream, addr)) => {
                    println!("🌐 新连接: {}", addr);

                    // 在协议层创建连接处理器
                    match protocol_server.create_connection() {
                        Ok((connection_id, handler)) => {
                            // 启动连接处理任务
                            Self::handle_connection(
                                connection_id,
                                stream,
                                handler,
                                Arc::clone(&connection_senders),
                                Arc::clone(&running),
                            ).await;
                        }
                        Err(e) => {
                            eprintln!("❌ 创建连接处理器失败: {}", e);
                        }
                    }
                }
                Err(e) => {
                    eprintln!("❌ 接受连接失败: {}", e);
                    sleep(Duration::from_millis(100)).await;
                }
            }
        }
    }

    /// 处理单个连接
    async fn handle_connection(
        connection_id: ConnectionId,
        stream: TcpStream,
        protocol_handler: DecoupledConnectionHandler,
        connection_senders: Arc<Mutex<HashMap<ConnectionId, tokio_mpsc::UnboundedSender<Vec<u8>>>>>,
        running: Arc<tokio::sync::RwLock<bool>>,
    ) {
        println!("🔄 开始处理连接 {}", connection_id);

        // 分离读写流
        let (mut read_half, mut write_half) = stream.into_split();

        // 创建发送器用于向连接写入数据
        let (write_sender, mut write_receiver) = tokio_mpsc::unbounded_channel();
        
        // 存储发送器
        {
            let mut senders = connection_senders.lock().unwrap();
            senders.insert(connection_id, write_sender);
        }

        // 使用Arc<Mutex<>>共享协议处理器
        let protocol_handler = Arc::new(Mutex::new(protocol_handler));

        // 启动读取任务
        let protocol_handler_read = Arc::clone(&protocol_handler);
        let running_read = Arc::clone(&running);
        let read_task = async move {
            let mut buffer = vec![0u8; 8192];
            
            while *running_read.read().await {
                match read_half.read(&mut buffer).await {
                    Ok(0) => {
                        println!("📡 连接 {} 已关闭", connection_id);
                        let handler = protocol_handler_read.lock().unwrap();
                        let _ = handler.handle_close();
                        break;
                    }
                    Ok(n) => {
                        // 将数据发送到协议层（同步）
                        let data = buffer[..n].to_vec();
                        let handler = protocol_handler_read.lock().unwrap();
                        if let Err(e) = handler.handle_data(data) {
                            eprintln!("❌ 连接 {} 发送数据到协议层失败: {}", connection_id, e);
                            break;
                        }
                    }
                    Err(e) => {
                        eprintln!("❌ 连接 {} 读取失败: {}", connection_id, e);
                        break;
                    }
                }
            }
        };

        // 启动写入任务
        let protocol_handler_write = Arc::clone(&protocol_handler);
        let running_write = Arc::clone(&running);
        let write_task = async move {
            while *running_write.read().await {
                tokio::select! {
                    // 处理来自网络层的写入请求
                    data = write_receiver.recv() => {
                        if let Some(data) = data {
                            if let Err(e) = write_half.write_all(&data).await {
                                eprintln!("❌ 连接 {} 写入失败: {}", connection_id, e);
                                break;
                            }
                            if let Err(e) = write_half.flush().await {
                                eprintln!("❌ 连接 {} flush失败: {}", connection_id, e);
                                break;
                            }
                        }
                    }
                    
                    // 处理来自协议层的响应
                    _ = sleep(Duration::from_millis(1)) => {
                        // 检查协议层的响应
                        loop {
                            let response = {
                                let handler = protocol_handler_write.lock().unwrap();
                                handler.try_get_response()
                            };
                            
                            if let Some(response) = response {
                                match response {
                                    ConnectionResponse::SendData(data) => {
                                        if let Err(e) = write_half.write_all(&data).await {
                                            eprintln!("❌ 连接 {} 写入协议响应失败: {}", connection_id, e);
                                            return;
                                        }
                                        if let Err(e) = write_half.flush().await {
                                            eprintln!("❌ 连接 {} flush失败: {}", connection_id, e);
                                            return;
                                        }
                                        println!("📤 连接 {} 发送响应: {} 字节", connection_id, data.len());
                                    }
                                    ConnectionResponse::Finished => {
                                        println!("✅ 连接 {} 协议处理完成", connection_id);
                                        return;
                                    }
                                    ConnectionResponse::Error(e) => {
                                        eprintln!("❌ 连接 {} 协议错误: {}", connection_id, e);
                                        return;
                                    }
                                }
                            } else {
                                break; // 没有更多响应，退出循环
                            }
                        }
                    }
                }
            }
        };

        // 并发运行读写任务
        tokio::select! {
            _ = read_task => {},
            _ = write_task => {},
        }

        // 清理连接
        {
            let mut senders = connection_senders.lock().unwrap();
            senders.remove(&connection_id);
        }

        // 停止协议处理器
        let mut handler = protocol_handler.lock().unwrap();
        if let Err(e) = handler.stop() {
            eprintln!("❌ 停止连接 {} 协议处理器失败: {}", connection_id, e);
        }

        println!("✅ 连接 {} 处理完成", connection_id);
    }

    /// 获取活跃连接数
    pub fn get_active_connections_count(&self) -> usize {
        self.connection_senders.lock().unwrap().len()
    }

    /// 获取协议层统计
    pub fn get_protocol_stats(&self) -> usize {
        self.protocol_server.get_active_connections_count()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use tokio::net::TcpStream;

    #[tokio::test]
    async fn test_tokio_network_server() {
        let storage_dispatcher = SingleThreadCommandDispatcher::new();
        let server = TokioNetworkServer::new(storage_dispatcher);

        // 启动服务器
        assert!(server.start("127.0.0.1:0").await.is_ok());

        // 等待启动
        sleep(Duration::from_millis(100)).await;

        // 停止服务器
        assert!(server.stop().await.is_ok());
    }

    #[tokio::test]
    async fn test_connection_isolation() {
        // 测试网络层和协议层的隔离
        let storage_dispatcher = SingleThreadCommandDispatcher::new();
        let _server = TokioNetworkServer::new(storage_dispatcher);

        // 协议层可以独立测试，不依赖Tokio运行时
        let protocol_server = DecoupledProtocolServer::new(SingleThreadCommandDispatcher::new());
        let (conn_id, mut handler) = protocol_server.create_connection().unwrap();

        // 在协议层处理命令（完全同步）
        let ping_data = b"*1\r\n$4\r\nPING\r\n".to_vec();
        assert!(handler.handle_data(ping_data).is_ok());

        // 等待处理
        sleep(Duration::from_millis(10)).await;

        // 检查响应
        if let Some(ConnectionResponse::SendData(response)) = handler.try_get_response() {
            let response_str = String::from_utf8_lossy(&response);
            assert!(response_str.contains("PONG"));
            println!("✅ 协议层独立处理成功: {}", response_str.trim());
        }

        assert!(handler.stop().is_ok());
        assert!(protocol_server.remove_connection(conn_id).is_ok());
    }
}